package falcon.protocol;

import java.util.HashMap;
import java.util.concurrent.ConcurrentLinkedQueue;

import falcon.common.Errno;
import falcon.common.Global;
import falcon.common.Log;
import falcon.protocol.PacketFactory.FieldBytes;
import falcon.protocol.PacketFactory.FieldInfo;
import falcon.protocol.PacketFactory.FieldType;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.PooledByteBufAllocator;

public final class Protocol {

	public static final int VERSION = 0x1;
	
	public Protocol() {
		m_field = PREPARE_PHASE;
		m_buffer = new byte[MAX_BYTES_LENGTH];
		m_queue = new ConcurrentLinkedQueue<Packet>();
	}
	
	public Packet get() {
		return m_queue.poll();
	}
	
	public int decode(ByteBuf buf) {
		int errno = Errno.SUCCESS;
		while (errno == Errno.SUCCESS) {
			if (m_field <= PREPARE_PHASE) {
				errno = preparePhase();
			} else if (m_field == PARSE_PHASE) {
				errno = parsePhase(buf);
			} else if (m_field < m_fieldCount) {
				errno = handlePhase(buf);
			} else {
				errno = completePhase();
			}
		}
		if (errno == Errno._NO_DATA_AVAILABLE) {
			errno = Errno.SUCCESS;
		}
		return errno;
	}
	
	public ByteBuf encode(Packet packet) {
		int cmd = packet.getCmd();
		FieldInfo[] fields = PacketFactory.getFields(cmd);
		if (fields == null) {
			Log.log("get fields failure");
			return null;
		}
		HashMap<String, Object> map = new HashMap<String, Object>();
		int total = fetch(fields, packet, map);
		if (total < 0) {
			Log.log("fetch field's value failure");
			return null;
		}
		if (total > MAX_PACKET_LENGTH) {
			Log.log("packet out of range;total=%d", total);
			return null;
		}
		packet.setBytes(total);
		return serialize(fields, packet, map);
	}
	
	private int preparePhase() {
		m_offset = 0;
		m_total = 0;
		m_read = 0;
		m_field = PARSE_PHASE;
		m_fieldCount = 1;
		return Errno.SUCCESS;
	}
	
	private int parsePhase(ByteBuf buf) {
		int errno = fillBuffer(buf, FieldBytes.FIX_INT);
		if (errno != Errno.SUCCESS) {
			return errno;
		}
		int header = decodeFixInt(m_buffer);
		int version = Packet.getVersion(header);
		if (version != Protocol.VERSION) {
			Log.log("");
			return Errno.PROTOCOL_VERSION_ERROR;
		}
		int cmd = Packet.getCmd(header);
		m_fields = PacketFactory.getFields(cmd);
		if (m_fields == null) {
			Log.log("");
			return Errno.PROTOCOL_SUCH_CMD;
		}
		m_packet = PacketFactory.create(cmd, header);
		if (m_packet == null) {
			Log.log("");
			return Errno._NO_SUCH_PROCESS;
		}
		m_field ++;
		m_fieldCount = m_fields.length;
		m_total = m_packet.getBytes();
		return Errno.SUCCESS;
	}
	
	private int handlePhase(ByteBuf buf) {
		FieldInfo info = m_fields[m_field];
		int bytes = info.bytes;
		if (bytes == FieldBytes.VAR_BYTES) {
			bytes = m_bytes;
		}
		String name = info.name;
		if (bytes > MAX_BYTES_LENGTH) {
			Log.log("field length out of range;name=%s", name);
			return Errno._VALUE_TOO_LARGE_FOR_DEFINED_DATA_TYPE;
		}
		int errno = fillBuffer(buf, bytes);
		if (errno != Errno.SUCCESS) {
			return errno;
		}
		if (m_read > m_total) {
			Log.log("packet read > total;name=%s;read=%d;total=%d", name, m_read, m_total);
			return Errno._VALUE_TOO_LARGE_FOR_DEFINED_DATA_TYPE;
		}
		m_offset = 0;
		FieldType type = info.type;
		if (type == FieldType.String && info.bytes == FieldBytes.FIX_SHORT) {
			m_bytes = decodeFixShort(m_buffer);
			m_field ++;
			return Errno.SUCCESS;
		}
		Object value;
		if (type == FieldType.String) {
			value = decodeString(m_buffer, bytes);
		} else if (type == FieldType.Int) {
			value = decodeFixInt(m_buffer);
		} else {// type == FieldType.Short
			value = decodeFixShort(m_buffer);
		}
		Log.log("value=%s", value);
		errno = m_packet.setFieldValue(name, value);
		if (errno != Errno.SUCCESS) {
			return errno;
		}
		m_field ++;
		return Errno.SUCCESS;
	}
	
	private int completePhase() {
		m_queue.add(m_packet);
		m_packet = null;
		m_field = PREPARE_PHASE;
		return Errno.SUCCESS;
	}
	
	private int fetch(FieldInfo[] fields, Packet packet, HashMap<String, Object> map) {
		int total = FieldBytes.FIX_INT;
		int count = fields.length;
		for (int i = 1; i < count; i ++) {
			FieldInfo info = fields[i];
			int bytes = info.bytes;
			if (bytes == FieldBytes.VAR_BYTES) {
				continue;
			}
			String name = info.name;
			Object value = packet.getFieldValue(name);
			if (value == null) {
				Log.log("get field value failure;field=%s", name);
				return -1;
			}
			if (info.type == FieldType.String) {
				if (!(value instanceof String)) {
					Log.log("field type error;name=%s", name);
					return -1;
				}
				String s = (String) value;
				byte[] a = s.getBytes(Global.DEFAULT_CHARSET);
				if (a.length > MAX_BYTES_LENGTH) {
					Log.log("field data out of range;name=%s;length=%d", name, a.length);
					return -1;
				}
				value = a;
				bytes += a.length;
			}
			map.put(info.name, value);
			total += bytes;
		}
		return total;
	}
	
	private ByteBuf serialize(FieldInfo[] fields, Packet packet, HashMap<String, Object> map) {
		int total = packet.getBytes();
		ByteBuf buf = PooledByteBufAllocator.DEFAULT.buffer(total);
		buf.writeInt(packet.header);
		int count = fields.length;
		for (int i = 1; i < count; i ++) {
			FieldInfo info = fields[i];
			if (info.bytes == FieldBytes.VAR_BYTES) {
				continue;
			}
			String name = info.name;
			FieldType type = info.type;
			Object obj = map.get(name);
			if (type == FieldType.String && obj instanceof byte[]) {
				byte[] array = (byte[]) obj;
				buf.writeShort(array.length);
				buf.writeBytes(array);
			} else if (type == FieldType.Int && obj instanceof Number) {
				Number num = (Number) obj;
				buf.writeInt(num.intValue());
			} else if (type == FieldType.Short && obj instanceof Number) {
				Number num = (Number) obj;
				buf.writeShort(num.shortValue());
			} else {
				Log.log("field type cast error;name=%s;type=%s", name, type);
				return null;
			}
		}
		return buf;
	}
	
	private int fillBuffer(ByteBuf buf, int bytes) {
		int remain = buf.readableBytes();
		if (remain + m_offset < bytes) {
			buf.readBytes(m_buffer, m_offset, remain);
			m_offset += remain;
			return Errno._NO_DATA_AVAILABLE;
		}
		int fill = bytes - m_offset;
		buf.readBytes(m_buffer, m_offset, fill);
		m_read += bytes;
		return Errno.SUCCESS;
	}
	
	private static int decodeFixInt(byte[] buffer) {
		int b3 = buffer[0] & 0xff;
		int b2 = buffer[1] & 0xff;
		int b1 = buffer[2] & 0xff;
		int b0 = buffer[3] & 0xff;
		return (b3 << 24) | (b2 << 16) | (b1 << 8) | b0;
	}
	
	private static int decodeFixShort(byte[] buffer) {
		int b1 = buffer[0] & 0xff;
		int b0 = buffer[1] & 0xff;
		return (b1 << 8) | b0;
	}
	
	private static String decodeString(byte[] buffer, int length) {
		return new String(buffer, 0, length, Global.DEFAULT_CHARSET);
	}
	
	private static final int PREPARE_PHASE = -1;
	private static final int PARSE_PHASE = 0;
	private static final int MAX_BYTES_LENGTH = 1 << 13;
	private static final int MAX_PACKET_LENGTH = 1 << 16;
	
	private int m_field;
	private int m_fieldCount;
	private int m_total;
	private int m_read;
	private int m_offset;
	private int m_bytes;
	private byte[] m_buffer;
	private Packet m_packet;
	private FieldInfo[] m_fields;
	private ConcurrentLinkedQueue<Packet> m_queue;
}
